package com.siebre.picclife.asynctask.service.support;

import java.util.Iterator;
import java.util.List;
import java.util.Queue;
import java.util.UUID;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.DataAccessException;
import org.springframework.transaction.annotation.Transactional;

import com.google.common.base.Predicate;
import com.google.common.collect.AbstractIterator;
import com.google.common.collect.FluentIterable;
import com.google.common.collect.Queues;
import com.siebre.picclife.asynctask.TaskException;
import com.siebre.picclife.asynctask.model.Task;
import com.siebre.picclife.asynctask.service.TaskRepository;
import com.siebre.picclife.asynctask.service.TaskService;

/**
 * 
 * @author ZhangChi
 * @since 2015年4月3日
 */
public class DefaultTaskService implements TaskService {
	
	private Logger logger = LoggerFactory.getLogger(getClass());
	
	private int maxRetryCount = 5;
	
	private String id;
	
	private Predicate<Task> corruptionPredicate;
	
	private TaskRepository repository;

	public DefaultTaskService(TaskRepository repository) {
		this.repository = repository;
		id = UUID.randomUUID().toString();
		corruptionPredicate = new Predicate<Task>() {

			public boolean apply(Task input) {
				return input.getFailureCount() >= maxRetryCount;
			}
		};
	}

	public String getId() {
		return id;
	}

	/**
	 * Specify max retry count for all tasks, defaults to 5
	 * <p>
	 * This limitation is applied before restarting a task.
	 * 
	 * @param maxRetryCount
	 */
	public void setMaxRetryCount(int maxRetryCount) {
		this.maxRetryCount = maxRetryCount;
	}

	@Transactional
	public Task startById(long id) {
		Task task = repository.findOne(id);
		
		boolean started;
		try {
			started = tryStartTask(task);
		} catch (RuntimeException e) {
			throw new TaskException("failed to start task, id=" + id, e);
		}
		
		if (!started) {
			throw new TaskException("failed to start task, id=" + id);
		}
		
		return task;
	}

	public List<Task> start(final int maxCount) {
		TaskQuery query = new TaskQuery() {
			
			public List<Task> execute(int maxCount) {
				return repository.findNewTasks(0, maxCount);
			}
		};
		return FluentIterable.from(asIterable(query, maxCount))
				.filter(new Predicate<Task>() {
		
					public boolean apply(Task task) {
						return tryStartTask(task);
					}
				}).limit(maxCount).toList();
	}

	private boolean tryStartTask(Task task) {
		if (corruptionPredicate.apply(task)) {
			try {
				markAsCorrupted(task);
			} catch (RuntimeException e) {
				//ignore optimistic locking failure
			}
			return false;
		}

		try {
			startTask(task);
			return true;
		} catch (RuntimeException e) {
			//ignore optimistic locking failure
			return false;
		}
	}

	private void markAsCorrupted(Task task) {
		task.markAsCorrupted();
		repository.save(task);
	}

	private void startTask(Task task) {
		task.startBy(id);
		repository.save(task);
		logger.debug("{} starts task {}", this, task.getOid());
	}

	@Transactional
	public void completeTask(long id) {
		repository.findOne(id).complete();
	}

	@Transactional
	public void completeTaskWithFailure(long id, Throwable cause) {
		repository.findOne(id).complete(cause);
	}

	public void completeTask(Task task) {
		task.complete();
		repository.save(task);
	}

	public void completeTaskWithFailure(Task task, Throwable cause) {
		task.complete(cause);
		repository.save(task);
	}

	public List<Task> restart(int maxCount) {
		TaskQuery query = new TaskQuery() {
			
			public List<Task> execute(int maxCount) {
				return repository.findFailedTasks(0, maxCount);
			}
		};
		return FluentIterable.from(asIterable(query, maxCount))
				.filter(new Predicate<Task>() {

					public boolean apply(Task task) {
						return tryStartTask(task);
					}
				}).limit(maxCount).toList();
	}

	public int recover(final String processor) {
		TaskQuery query = new TaskQuery() {
			
			public List<Task> execute(int maxCount) {
				return repository.findProcessingTasks(processor, 0, maxCount);
			}
		};
		return FluentIterable.from(asIterable(query, 10, Integer.MAX_VALUE))
				.filter(new Predicate<Task>() {

					public boolean apply(Task input) {
						//TODO check if it is stuck
						return true;
					}
				})
				.filter(new Predicate<Task>() {

					public boolean apply(Task task) {
						return fail(task);
					}
				})
				.size();
	}

	private boolean fail(Task task) {
		try {
			task.completeWithFailure("failed manually");
		} catch (IllegalStateException e) {
			return false;
		}
		try {
			repository.save(task);
			logger.debug("{} fails task {}", this, task.getOid());
			return true;
		} catch (DataAccessException e) {
			return false;
		}
	}

	private Iterable<Task> asIterable(TaskQuery query, int fetchSize) {
		return asIterable(query, fetchSize, 5);
	}

	private Iterable<Task> asIterable(final TaskQuery query, final int fetchSize, final int maxQueryCount) {
		return new Iterable<Task>() {

			public Iterator<Task> iterator() {
				return new AbstractIterator<Task>() {
					
					private Queue<Task> buffer = Queues.newLinkedBlockingQueue();
					
					private int queryCount = 0;

					@Override
					protected Task computeNext() {
						if (buffer.isEmpty()) {
							int limit = fetchSize * (queryCount + 1);
							if (queryCount >= maxQueryCount) {
								return endOfData();
							}
							//TODO abort query if limit becomes too large
							List<Task> queryResult = query.execute(limit);
							queryCount++;
							if (queryResult.isEmpty()) {
								return endOfData();
							}
							buffer.addAll(queryResult);
						}
						return buffer.poll();
					}
				};
			}
		};
	}

	private static interface TaskQuery {
		
		List<Task> execute(int maxCount);
	}
}
